#include <sys/socket.h>
#include <errno.h>
#include <netinet/in.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <sys/poll.h>
#include <sys/epoll.h>
#include <sys/time.h>
#include "reactor.h"
#include "kvs.h"

#define CONNECTION_SIZE 1000 // 1048576 // 1024 * 1024
#define PORT_COUNT 1		 // 100
#define MAX_PORTS 20

#define MAX_CMD_TOKEN_NUM 100
#define USE_SELF_TOKEN 0

int accept_cb(int fd);
int recv_cb(int fd);
int send_cb(int fd);
int init_server(int sockfd, int port);

conn *conn_list;

int accept_cb(int fd)
{
	struct sockaddr_in clientaddr;
	socklen_t len = sizeof(clientaddr);

	int clientfd = accept(fd, (struct sockaddr *)&clientaddr, &len);
	printf("clientfd:%d\n", clientfd);

	conn_register(clientfd, recv_cb, send_cb);
	set_event(clientfd, EPOLLIN | EPOLLET, 0);

	setNonblock(clientfd); // 设置非阻塞

	return 0;
}

int split_by_crlf(char *msg, char *tokens[])
{

	int idx = 0;
	char *token = strtok(msg, "\x03");

	while (token != NULL)
	{
		// LOG("idx[%d]: %s\n", idx, token);
		tokens[idx++] = token;
		token = strtok(NULL, "\x03");
	}
	return idx;
}

int recv_cb(int fd)
{

	char buffer[BUFFER_LENGTH] = {0};
	// 双向链表接收
	while (1)
	{

		int ret = recv(fd, buffer, BUFFER_LENGTH - 1, 0);
		buffer[ret] = '\0';
		if (ret == 0)
		{
			printf("client disconnect: %d\n", fd);
			conn_close(fd);
			return 0;
		}
		else if (ret < 0)
		{
			if (errno == EAGAIN || errno == EWOULDBLOCK)
			{
				// 非阻塞下暂时无数据，返回已读字节数
				break;
			}
			else
			{
				// 真正的错误（如连接重置、中断）
				printf("recv errno: %d --> %s\n", errno, strerror(errno));
				conn_close(fd);
				return -1;
			}
		}
		else
		{

			int flag = 0; // 最后一个分段是否为完整指令的标志
			if (buffer[ret - 1] == '\x03')
			{
				flag = 1;
			}

			if (buffer[0] == '\x03')
			{
				// 前一个接收末尾就是完整数据，插入即可
				if (!list_empty(&conn_list[fd].remain_link))
				{
					int total_len = conn_list[fd].remain_link.length + conn_list[fd].remain_len + 1;
					char *buf = kvs_malloc(total_len);
					memset(buf, 0, total_len);

					int len = fill_with_list(buf, conn_list[fd].remain_link.length, &conn_list[fd].remain_link);
					memcpy(buf + len, conn_list[fd].remain_buffer, conn_list[fd].remain_len);
					insert_list(buf, total_len, &conn_list[fd].rlink);

					clear_list(&conn_list[fd].remain_link);
					conn_list[fd].remain_len = 0;
					memset(conn_list[fd].remain_buffer, 0, BUFFER_LENGTH);
					kvs_free(buf);
				}

				else
				{
					insert_list(conn_list[fd].remain_buffer, conn_list[fd].remain_len + 1, &conn_list[fd].rlink);

					conn_list[fd].remain_len = 0;
					memset(conn_list[fd].remain_buffer, 0, BUFFER_LENGTH);
				}
			}

			char *tokens[MAX_CMD_TOKEN_NUM] = {NULL};
			int count = split_by_crlf(buffer, tokens);

			if (count == 0)
			{
				LOG("count == 0\n");
				continue;
			}

			for (int i = 0; i < count - 1; i++)
			{

				if (i == 0 && conn_list[fd].remain_len != 0)
				{
					if (!list_empty(&conn_list[fd].remain_link))
					{
						int total_len = conn_list[fd].remain_link.length + conn_list[fd].remain_len + strlen(tokens[i]) + 1;
						char *buf = kvs_malloc(total_len);
						memset(buf, 0, total_len);

						int len = fill_with_list(buf, conn_list[fd].remain_link.length, &conn_list[fd].remain_link);
						memcpy(buf + len, conn_list[fd].remain_buffer, conn_list[fd].remain_len);
						memcpy(buf + len + conn_list[fd].remain_len, tokens[i], strlen(tokens[i]) + 1);
						insert_list(buf, total_len, &conn_list[fd].rlink);

						clear_list(&conn_list[fd].remain_link);
						conn_list[fd].remain_len = 0;
						memset(conn_list[fd].remain_buffer, 0, BUFFER_LENGTH);
						kvs_free(buf);

						continue;
					}

					else
					{

						if (conn_list[fd].remain_len + strlen(tokens[i]) + 1 > BUFFER_LENGTH)
						{
							char *buf = kvs_malloc(conn_list[fd].remain_len + strlen(tokens[i]) + 1);
							memset(buf, 0, conn_list[fd].remain_len + strlen(tokens[i]) + 1);
							memcpy(buf, conn_list[fd].remain_buffer, conn_list[fd].remain_len);
							memcpy(buf + conn_list[fd].remain_len, tokens[i], strlen(tokens[i]) + 1);
							insert_list(buf, conn_list[fd].remain_len + strlen(tokens[i]) + 1, &conn_list[fd].rlink);

							conn_list[fd].remain_len = 0;
							memset(conn_list[fd].remain_buffer, 0, BUFFER_LENGTH);
							kvs_free(buf);

							continue;
						}

						else
						{

							memcpy(conn_list[fd].remain_buffer + conn_list[fd].remain_len, tokens[i], strlen(tokens[i]) + 1);
							insert_list(conn_list[fd].remain_buffer, conn_list[fd].remain_len + strlen(tokens[i]) + 1, &conn_list[fd].rlink);

							conn_list[fd].remain_len = 0;
							memset(conn_list[fd].remain_buffer, 0, BUFFER_LENGTH);
							continue;
						}
					}
				}

				insert_list(tokens[i], strlen(tokens[i]) + 1, &conn_list[fd].rlink);
			}

			if (flag == 1)
			{
				if (conn_list[fd].remain_len != 0)
				{
					if (!list_empty(&conn_list[fd].remain_link))
					{

						// LOG("remain_link:\n");
						// print_list(&conn_list[fd].remain_link);

						int total_len = conn_list[fd].remain_link.length + conn_list[fd].remain_len + strlen(tokens[count - 1]) + 1;

						// LOG("total_len: %d, remain_link_len: %d, remain_buf_len: %d, token_len: %ld\n",
						//	total_len, conn_list[fd].remain_link.length, conn_list[fd].remain_len, strlen(tokens[count - 1]));

						char *buf = kvs_malloc(total_len);
						memset(buf, 0, total_len);

						int len = fill_with_list(buf, conn_list[fd].remain_link.length, &conn_list[fd].remain_link);

						// LOG("buf1: %s\n", buf);

						memcpy(buf + len, conn_list[fd].remain_buffer, conn_list[fd].remain_len);

						// LOG("buf2: %s\n", buf);

						memcpy(buf + len + conn_list[fd].remain_len, tokens[count - 1], strlen(tokens[count - 1]) + 1);

						// LOG("buf3: %s\n", buf);

						insert_list(buf, total_len, &conn_list[fd].rlink);

						clear_list(&conn_list[fd].remain_link);
						conn_list[fd].remain_len = 0;
						memset(conn_list[fd].remain_buffer, 0, BUFFER_LENGTH);
						kvs_free(buf);
					}

					else
					{

						if (conn_list[fd].remain_len + strlen(tokens[count - 1]) + 1 > BUFFER_LENGTH)
						{
							char *buf = kvs_malloc(conn_list[fd].remain_len + strlen(tokens[count - 1]) + 1);
							memset(buf, 0, conn_list[fd].remain_len + strlen(tokens[count - 1]) + 1);
							memcpy(buf, conn_list[fd].remain_buffer, conn_list[fd].remain_len);
							memcpy(buf + conn_list[fd].remain_len, tokens[count - 1], strlen(tokens[count - 1]) + 1);
							insert_list(buf, conn_list[fd].remain_len + strlen(tokens[count - 1]) + 1, &conn_list[fd].rlink);

							conn_list[fd].remain_len = 0;
							memset(conn_list[fd].remain_buffer, 0, BUFFER_LENGTH);
							kvs_free(buf);
						}

						else
						{
							memcpy(conn_list[fd].remain_buffer + conn_list[fd].remain_len, tokens[count - 1], strlen(tokens[count - 1]) + 1);
							insert_list(conn_list[fd].remain_buffer, conn_list[fd].remain_len + strlen(tokens[count - 1]) + 1, &conn_list[fd].rlink);

							conn_list[fd].remain_len = 0;
							memset(conn_list[fd].remain_buffer, 0, BUFFER_LENGTH);
						}
					}
				}
				else
				{
					insert_list(tokens[count - 1], strlen(tokens[count - 1]) + 1, &conn_list[fd].rlink);
				}
			}

			else if (flag == 0)
			{
				int last_len = strlen(tokens[count - 1]);
				// LOG("conn_list[fd].remain_len + last_len = %d, BUFFER_LENGTH = %d\n", conn_list[fd].remain_len + last_len, BUFFER_LENGTH);
				if (conn_list[fd].remain_len + last_len + 1 > BUFFER_LENGTH)
				{
					insert_list(conn_list[fd].remain_buffer, conn_list[fd].remain_len, &conn_list[fd].remain_link);
					memset(conn_list[fd].remain_buffer, 0, BUFFER_LENGTH);
					memcpy(conn_list[fd].remain_buffer, tokens[count - 1], last_len);
					conn_list[fd].remain_len = last_len;
				}
				else
				{
					memcpy(conn_list[fd].remain_buffer + conn_list[fd].remain_len, tokens[count - 1], last_len);
					conn_list[fd].remain_len += strlen(tokens[count - 1]);
				}

				count = count - 1;
			}
			// LOG("符合格式的count: %d\n", count);
			// LOG("remain_len:%d, remain_buffer: %.*s\n", conn_list[fd].remain_len, conn_list[fd].remain_len, conn_list[fd].remain_buffer);
		}
	}

	// LOG("rlink: \n");
	// print_list(&conn_list[fd].rlink);
	kvs_request(&conn_list[fd].rlink, &conn_list[fd].wlink);
	clear_list(&conn_list[fd].rlink);

	set_event(fd, EPOLLOUT | EPOLLET, 1);
}

int send_cb(int fd)
{
	if (list_empty(&conn_list[fd].wlink))
	{
		// 防御：无数据可发送时直接切换回读事件
		set_event(fd, EPOLLIN | EPOLLET, 1);
		return 0;
	}

	// print_list(&conn_list[fd].wlink);
	int sent = send_list(fd, &conn_list[fd].wlink);
	if (sent < 0)
	{
		// 发送过程中出错，连接已关闭
		conn_close(fd);
		return -1;
	}

	if (!list_empty(&conn_list[fd].wlink))
	{
		// 还有数据未发送完，继续监听写事件
		set_event(fd, EPOLLOUT | EPOLLET, 1);
		return 0;
	}

	else
	{
		// printf("sendLen: %d\n", conn_list[fd].sent);
		conn_list[fd].sent = 0;
		clear_list(&conn_list[fd].wlink);

		set_event(fd, EPOLLIN | EPOLLET, 1);
		return 0;
	}
}




int reactor_entry(int port)
{

	conn_list = get_connlist(CONNECTION_SIZE);
	int epfd = init_epfd();
	LOG("epfd: %d\n", epfd);


	for (int i = 0; i < PORT_COUNT; i++)
	{
		int socketfd = socket(AF_INET, SOCK_STREAM, 0);

		init_server(socketfd, port + i);
		set_event(socketfd, EPOLLIN, 0);
		conn_register(socketfd, accept_cb, NULL);
	}

	struct epoll_event *events = kvs_malloc(sizeof(struct epoll_event) * CONNECTION_SIZE);
	memset(events, 0, sizeof(struct epoll_event) * CONNECTION_SIZE);
	while (1)
	{

		int nready = epoll_wait(epfd, events, CONNECTION_SIZE, -1);

		for (int i = 0; i < nready; i++)
		{
			int fd = events[i].data.fd;
			if (events[i].events & EPOLLIN)
			{
				conn_list[fd].in_callback(fd);
			}

			if (events[i].events & EPOLLOUT)
			{
				conn_list[fd].out_callback(fd);
			}
		}
	}

	close(epfd);
	kvs_free(events);
	close_connlist(conn_list, CONNECTION_SIZE);
	return 0;
}

int init_server(int sockfd, int port)
{
	struct sockaddr_in servaddr;
	servaddr.sin_family = AF_INET;
	servaddr.sin_port = htons(port);
	servaddr.sin_addr.s_addr = htonl(INADDR_ANY);

	setReUseAddr(sockfd); // 启用端口复用，注意一定要放在bind前面。否则端口复用无效，程序断开重启后，bind会失效

	bind(sockfd, (struct sockaddr *)&servaddr, sizeof(servaddr));
	listen(sockfd, 5);
	LOG("listen at: %d\n", port);
	LOG("listenfd: %d\n", sockfd);
}
